package com.comtop.db.fastar

import java.util.Arrays
import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD
import scala.collection.mutable.ArraySeq
import com.comtop.db.dm.JdbcConnction
import java.text.SimpleDateFormat
import java.sql.Timestamp

object Test {
 
   def writeDataCount(){
     val sc = JdbcConnection.sc
     val input = sc.textFile("E:/backup/data/fData.txt", 1)
     val words = input.flatMap( line => line.split(","))
     val counts = words.map(word => (word,1)).reduceByKey((x,y)=>x+y)
//     println("[统计单词个数]: " + counts.countByValue())
     counts.countByValue().take(100).foreach(x => println(x))
   } 
  
  
   def testMKstring(){
     val rdd = JdbcConnection.sc.parallelize(List(1,2,3,4,5,6,7,8,9), 1)
     val r = rdd.filter { x => x > 3 }
     println(">3 : "+r.collect().mkString(","))
     
     println("max - min = " + (r.max() - r.min()))
     println("sum = "+r.sum())
   }
   
   
   /**
   * 将时间字符串转换为Timestamp格式
   */
   def getTimestamp(x:String) :java.sql.Timestamp = {
        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        
        try {  
            if (null == x || x == "")
              return null
            else {
              val d = format.parse(x);
              val t = new Timestamp(d.getTime());
              return t
            }
        } catch {  
          case e: Exception => println("cdr parse timestamp wrong")
         } 
        return null
    }

   
   /**
    * 展示结果
    */
  def showMyFruit(fruit: String) {
    if(fruit.equals("pears")){
    	println("I like " + fruit)
    }else{
     	println("I don't like " + fruit)
    }
  }
   
   def countPowerLineLoss(): Unit = {
     val fruit: List[String] = List("apples", "oranges", "pears")
     val rdd = JdbcConnection.sc.parallelize(fruit)
     rdd.map{x=>showMyFruit(x)}.collect()
   }
   
   def main(args: Array[String]): Unit = {
     
    /**
     * 序列化数组=>RDD
     */
//    val data = ArraySeq(1,2,3,4,5)
//    val a = JdbcConnection.sc.parallelize(data, 3)
//    val rdd = a.collect()
//    println("[统计个数]: " + rdd.length)
    
    
//    JdbcConnection.getFilePath
//    writeDataCount
    
    
//    val fastar = new FastarUtils 
//    fastar.getDataFromFastar()
    
     
//     writeDataCount
     
//     testMKstring
     
//     println("转换后的时间戳格为："+getTimestamp("2017-03-31 14:43:21:000").toString().subSequence(0, 13))
     
     
//     val timeFormat = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
//     val startTime:Long = timeFormat.parse("2016-06-01 14:43:21").getTime()
//     val endTime:Long =  timeFormat.parse("2016-06-05 14:43:21").getDate+1
//     
//     println("startTime="+startTime+"  "+"endTime="+endTime)
     
//     showMyFruit("apple")
     countPowerLineLoss
   }
   
   
}